package com.atguigu.flink.state;

import com.atguigu.flink.state.opearatestate.Flink02_ListState;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * Created by 黄凯 on 2023/6/22 0022 20:53
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 状态后端。 StateBackend
 * *      作用： 管理(读，写，备份)状态。
 * *
 * *      HashMapStateBackend： 默认。
 * *              特点： 在TaskManager的内存中，使用HashMap存储所有的状态信息。
 * *                      好处：读写速度快。
 * *                      弊端： 内存是有限的，有大的状态，存不了！
 * *
 * *              设置： 不用，默认。
 * *                  配置文件中设置:   state.backend.type: hashmap
 * *
 * *      rocksDBStateBackend： 可以配置。用的最多的。
 * *              配置: state.backend.type: rocksdb
 * *
 * *              rocksDB是flink开发的一个内嵌式数据库(derby)。
 * *                  这个数据库无需额外安装，是随着Java启动而生成。
 * *
 * *              特点： 可以把状态存储到磁盘！
 * *                      好处: 存储大状态，无限存储！ 生产用。
 * *                      弊端:  性能低，比HashMapStateBackend慢10倍。
 * *                               磁盘IO
 * *                               格式： 字节数组。
 * *                               存储序列化和反序列化的开销。
 * *
 * *     -------------------------------------
 * *      状态通过状态后端备份到哪里了?
 * *              checkpoint 指定备份的设备(位置)
 * *                  默认： JobManagerCheckpointStorage 不可靠。
 * *
 * *                  选择备份到文件系统是可靠的方式。FileSystemCheckpointStorage
 * *                      本地
 * *                      HDFS
 */
public class Flink13_StateBackend {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

        //代码中设置 HashMapStateBackend
        HashMapStateBackend hashMapStateBackend = new HashMapStateBackend();
        env.setStateBackend(hashMapStateBackend);

        //代码设置rocksDBStateBackend
        EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend();
        env.setStateBackend(rocksDBStateBackend);

        //设置ck为本地文件系统
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
//        checkpointConfig.setCheckpointStorage("file:///E:/ck");
//        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck");

        //开启了checkpoint（备份），每间隔2s自动备份，程序就会无限次重启
        env.enableCheckpointing(2000);

        env
                .socketTextStream("127.0.0.1", 8888)
                .map(new MyMapFunction())
                .addSink(new SinkFunction<String>() {
                    @Override
                    public void invoke(String value, Context context) throws Exception {
                        if (value.contains("x")) {
                            //人为手动抛出一个异常，为了模拟故障的情况
                            throw new RuntimeException("出异常了!");
                        }
                        System.out.println(value);
                    }
                });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    /*
        功能： 实现字符串的累积打印

            Flink提供的状态就是拥有自动备份，自动恢复功能的集合。
     */
    public static class MyMapFunction implements MapFunction<String, String>, CheckpointedFunction {

        //当集合用，本质是ManagedState中的OpearateState
        ListState<String> strs;
        ListState<String> strs1;

        @Override
        public String map(String s) throws Exception {
            //存
            strs.add(s);
            //取
            return strs.get().toString();
        }

        /*
            周期性(默认200ms，可以调整)将状态以快照的形式进行备份
         */
        @Override
        public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
            System.out.println("MyMapFunction.snapshotState");
            //备份是自动进行的，无需进行任何手动操作
        }

        /*
            初始化： 在第一次启动或任务失败重启后执行，从之前的备份中找到状态，重新赋值。

            FunctionInitializationContext context: 程序的运行环境，或上下文，可以从中获取很多额外的信息。
         */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyMapFunction.initializeState");
            //找到备份状态的存储设备
            OperatorStateStore operatorStateStore = context.getOperatorStateStore();
            //从备份中找到之前备份的变量
            ListStateDescriptor<String> listStateDescriptor1 = new ListStateDescriptor<>("list1", String.class);
            ListStateDescriptor<String> listStateDescriptor2 = new ListStateDescriptor<>("list2", String.class);

            strs1 = operatorStateStore.getListState(listStateDescriptor2);
            strs = operatorStateStore.getListState(listStateDescriptor1);
        }
    }

}
